package com.bjtu.boatmq.client.producer;

import java.util.ArrayList;
import java.util.List;

import com.bjtu.boatmq.common.log.Log;
import com.bjtu.boatmq.common.protocol.Message;
import com.bjtu.boatmq.common.protocol.PacketSource;
import com.bjtu.boatmq.common.protocol.RequestCode;
import com.bjtu.boatmq.common.protocol.ResponseCode;
import com.bjtu.boatmq.common.protocol.SourceDecoder;
import com.bjtu.boatmq.common.protocol.SourceEncoder;

import cn.boat.router.models.Broker;
import cn.boat.router.models.Topic;
import io.netty.bootstrap.Bootstrap;
import io.netty.channel.Channel;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.SimpleChannelInboundHandler;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.nio.NioSocketChannel;

public class SendMsg {
	private Bootstrap bootstrap;
    private NioEventLoopGroup group;
    private ChannelFuture future;
    private Log log;
    public SendMsg() {
        log=new Log();
    	bootstrap = new Bootstrap();
    	group = new NioEventLoopGroup();
    	bootstrap.group(group)
        .channel(NioSocketChannel.class)
        .handler(new ChannelInitializer<Channel>() {
            @Override
            protected void initChannel(Channel ch) {
                ch.pipeline().addLast(new SourceEncoder());
                ch.pipeline().addLast(new SourceDecoder());
                ch.pipeline().addLast(new SimpleChannelInboundHandler<PacketSource>() {
                	Log l=new Log();
                    @Override
                    protected void channelRead0(ChannelHandlerContext ctx, PacketSource ps) {
                    	//接收结果处理
                    	if(ps.getCode()==ResponseCode.NO_ERROR) {
                    		l.writeInfo("Sent successfully:REQUEST--"+ps.getRequestId());
                    	}
                    	else {
                    		l.writeError("Sent Error------Error code:"+ps.getCode());
                    	}
                    }
                });
            }
        });
    }
    public PacketSource getPacketSource() {
    	PacketSource packetSource=new PacketSource();
    	packetSource=new PacketSource();
        packetSource.setCode(RequestCode.SEND_MESSAGE);
        packetSource.setFlag(PacketSource.REQUEST);
        packetSource.setSponsorType(PacketSource.PRODUCER);
        return packetSource;
    }
    public void send(String addr,int port,PacketSource ps) {
    	future=bootstrap.connect(addr, port);
    	future.channel().writeAndFlush(ps);
    }
    public void send(List<Message> m,Router r) throws Exception {
    	this.log.writeInfo("send a list of msg");
    	if(!m.isEmpty()) {
    		Topic topic= r.getTopic(m.get(0).getTopic());
    		if(topic!=null) {
    			if(m.get(0).getPartitionId()==-1) {
    				int partion=Selector.select(topic.getPartitionNum());
    				for(Message msg:m)
    					msg.setPartitionId(partion);
    			}
    			int brokerid=topic.getPartitionList().get(m.get(0).getPartitionId()).getLeader();
    			Broker broker=r.getBrokerById(brokerid);
    			PacketSource packetSource=this.getPacketSource();
    			packetSource.setData(m);
    			this.send(broker.getBrokerHost(), broker.getBrokerPort(), packetSource);
    		}
    		else {
    			this.log.writeError("The Topic does not exist!!!");
    			r.getRoute(m.get(0).getTopic());
    		}
    	}
    	
    }
    
}
